package com.gmall.controller;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.gmall.utils.Constants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 *
 * @author Demik Qin
 * @date 2020/10/14
 */
@RestController
public class LoggerController {
    @Autowired
    private KafkaTemplate<String, String> KafkaTemplate;

    private final Logger logger = LoggerFactory.getLogger(LoggerController.class);


    @PostMapping("/log")
    public String doLog(@RequestParam("log") String log) {
        // 1、给日志数据添加时间戳
        log = addTS(log);
        // 2、把数据落盘 (将来离线处理可以使用)
        saveToDisk(log);
        // 3、直接写入到kafka  1.得到生产者 2.通过生产者向kafka写数据
        sendToKafka(log);
        System.out.println(log);
        return "success";
    }


    /**
     * 将日志发送到kafka
     *
     * @param log 日志
     */
    private void sendToKafka(String log) {
       /* String topic = Constant.EVENT_TOPIC;
        if (log.contains("startup")) {
            topic = Constant.STARTUP_TOPIC;
        }*/
        // 不同的日志, 写到不同的topic中
        String topic = log.contains("startup") ? Constants.STARTUP_TOPIC : Constants.EVENT_TOPIC;
        KafkaTemplate.send(topic, log);
    }


    /**
     * 将日志写入到磁盘
     *
     * @param log 日志
     */
    private void saveToDisk(String log) {
        logger.warn(log);
        logger.error(log);
    }

    /**
     * 给日志添加时间戳
     *
     * @param log 日志
     * @return 带时间戳的日志
     */
    private String addTS(String log) {
        JSONObject obj = JSON.parseObject(log);
        obj.put("ts", System.currentTimeMillis());
        return obj.toJSONString();
    }

}
